package com.hd.determine.config.rabbitmq.consumer;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.hd.determine.config.rabbitmq.MqQueueConstant;
import com.hd.determine.config.rabbitmq.batch.AlarmStream;
import com.hd.determine.config.rabbitmq.model.AlarmMQ;
import com.hd.determine.modules.alarm.determine.mediator.DetermineMediator;
import com.hd.determine.modules.alarm.model.AlarmDetail;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.Objects;

/**
 * @author chentl
 * @version V1.0
 * @Project 交通运输第三方监测平台
 * @Title RabbitConsumer.java
 * @Description rabbitMQ 消费者
 * @Package com.hd.determine.config.rabbitmq.consumer
 * @date 2021/1/6 3:36 下午
 * @Copyright: 深圳市宏电技术股份有限公司 All rights reserved.
 */
@Slf4j
@Component
public class RabbitConsumer {


    @Resource
    private DetermineMediator determineMediator;

    @Resource
    private AlarmStream alarmStream;


    /**
     * @return void
     * @author chentl
     * @description 报警详情队列消费者
     * @date 2021/1/6 3:54 下午
     * @params [bt]
     */
    @RabbitListener(queues = MqQueueConstant.ALARM_DETAIL_QUEUE, containerFactory = "alarmContainerFactory")
    @RabbitHandler
    public void alarmDetailQueue(Message msg, Channel channel) throws IOException {
        String message = new String(msg.getBody());
        try {
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);

            determineMediator.prepareDetermine(message);
        } catch (Exception e) {
            channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, false);
            log.error("[rabbit] 报警处理队列发生异常", e);
            log.error("[rabbit] 异常原始消息: {}", message);
        }

    }


    /**
     * @return void
     * @author chentl
     * @description 报警数据 生成 或更新队列
     * @date 2021/3/15 3:16 下午
     * @params [msg, channel]
     */
    @RabbitListener(queues = MqQueueConstant.ALARM_GENERATE_QUEUE, containerFactory = "alarmGenContainerFactory")
    @RabbitHandler
    public void alarmGenerateQueue(@Payload AlarmMQ alarmMQ, Message msg, Channel channel) throws IOException {
        try {
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);

            AlarmDetail alarmDetail = alarmMQ.alarmDetail();
            if (Objects.isNull(alarmDetail)) {
                log.error("[determine] 报警对象为 null，跳过");
                return;
            }

            alarmStream.streamInsert(alarmDetail);


        } catch (Exception e) {
            channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, false);
            log.error("[rabbit] 报警数据生成队列发生异常", e);
            log.error("[rabbit] 异常原始消息: {}", JSON.toJSONString(alarmMQ));
        }

    }

    @RabbitListener(queues = MqQueueConstant.ALARM_UPDATE_QUEUE, containerFactory = "alarmGenContainerFactory")
    @RabbitHandler
    public void alarmUpdateQueue(@Payload AlarmMQ alarmMQ, Message msg, Channel channel) throws IOException {
        try {
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);

            AlarmDetail alarmDetail = alarmMQ.alarmDetail();

            if (Objects.isNull(alarmDetail)) {
                log.error("[determine] 报警对象为 null，跳过");
                return;
            }

            alarmStream.streamUpdate(alarmDetail);

        } catch (Exception e) {
            channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, false);
            log.error("[rabbit] 报警数据更新队列发生异常", e);
            log.error("[rabbit] 异常原始消息: {}", JSON.toJSONString(alarmMQ));
        }
    }


}
